package com.pipi.mq.rocketmq.base.consumer;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.UUID;

/**
 * All rights Reserved, Designed By www.maihaoche.com
 *
 * 因需依赖高版本的camaro
 * 故将 camaro-api:0.0.5中的rocketmq中的代码复制出来，解决兼容问题
 * @Package com.golf.core.rocketmq
 * @author: 皮皮（pipi@maihaoche.com）
 * @date: 2019/5/15 10:19 AM
 * @Copyright: 2017-2020 www.maihaoche.com Inc. All rights reserved.
 * 注意：本内容仅限于卖好车内部传阅，禁止外泄以及用于其他的商业目
 */
public abstract class AbstractMQConsumer<T> {

    @Autowired
    private DefaultMQPushConsumer consumer;

    private String topic;
    private String consumerGroup;
    private String messageMode = MessageModel.CLUSTERING.getModeCN();
    private String consumeMode = "CONCURRENTLY";
    private String tag;

    public AbstractMQConsumer() {
    }

    public String getConsumeMode() {
        return this.consumeMode;
    }

    public String getMessageMode() {
        return this.messageMode;
    }

    public void setMessageModel(MessageModel messageModel) {
        consumer.setMessageModel(messageModel);
    }

    public String getTopic() {
        return this.topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getConsumerGroup() {
        return this.consumerGroup;
    }

    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

    public String getTag() {
        return this.tag;
    }

    public void setTag(String tag) {
        this.tag = tag;
    }

    @PostConstruct
    public void init() {
        try {
            consumer.setConsumerGroup(this.getConsumerGroup());
            consumer.subscribe(this.getTopic(), this.getTag());
            consumer.setInstanceName(UUID.randomUUID().toString());
            consumer.setMessageModel(MessageModel.valueOf(this.getMessageMode()));
            if (this.getConsumeMode().equals("CONCURRENTLY")) {
                consumer.registerMessageListener((MessageListenerConcurrently) this::dealMessage);
            } else {
                if (!this.getConsumeMode().equals("ORDERLY")) {
                    throw new RuntimeException("unknown consume mode ! only support CONCURRENTLY and ORDERLY");
                }

                consumer.registerMessageListener((MessageListenerOrderly) this::dealMessage);
            }

            consumer.start();
        } catch (Exception e) {
            throw new RuntimeException("消息初始化失败, 异常:{}" + e.getMessage(), e);
        }
    }

    public abstract boolean process(T var1);

    public ConsumeConcurrentlyStatus dealMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        Iterator var3 = list.iterator();

        MessageExt messageExt;
        Object t;
        do {
            if (!var3.hasNext()) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }

            messageExt = (MessageExt)var3.next();
            if (messageExt.getReconsumeTimes() != 0) {
            }

            t = this.parseMessage(messageExt);
        } while(null == t || this.process((T) t));

        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }

    public ConsumeOrderlyStatus dealMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        Iterator var3 = list.iterator();

        MessageExt messageExt;
        Object t;
        do {
            if (!var3.hasNext()) {
                return ConsumeOrderlyStatus.SUCCESS;
            }

            messageExt = (MessageExt)var3.next();
            if (messageExt.getReconsumeTimes() != 0) {
            }

            t = this.parseMessage(messageExt);
        } while(null == t || this.process((T) t));
        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
    }

    private T parseMessage(MessageExt message) {
        if (message != null && message.getBody() != null) {
            Type type = this.getMessageType();
            if (type instanceof Class) {
                try {
//                    Object data = JSON.parseObject(new String(message.getBody()), type);
                    Object data = new String(message.getBody());
                    return (T) data;
                } catch (Exception var4) {
                    var4.printStackTrace();
                }
            } else {
            }

            return null;
        } else {
            return null;
        }
    }

    private Type getMessageType() {
        Type superType = this.getClass().getGenericSuperclass();
        return (Type)(superType instanceof ParameterizedType ? ((ParameterizedType)superType).getActualTypeArguments()[0] : Object.class);
    }
}
